package com.liyu.rabbit;

import com.liyu.rabbit.factory.RabbitFactory;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.MessageProperties;

import java.nio.charset.StandardCharsets;

/**
 * @author 李誉
 * @date 2022-05-18 22:29
 * @description 发布确认
 */
public class PublisherConfirm {

    public static void main(String[] args) throws Exception {
        singleConfirm(); // 单个发布确认模式，发送1000条消息，消耗时间1787ms
        batchConfirm(); // 批量发布确认模式，发送1000条消息，消耗时间95ms
        asyncConfirm(); // 异步发布确认模式，发送1000条消息，消耗时间47ms
    }

    /**
     * 单个发布确认
     *  只有一条消息确认后，才能发新的消息
     */
    public static void singleConfirm () throws Exception {

        // 创建信道
        Channel channel = RabbitFactory.createNewChannel();

        // 开启发布确认
        channel.confirmSelect();

        /*
         * 创建一个队列
         * 参数：
         *  queue：队列名称
         *  durable：是否持久化(默认存储在内存中)
         *  exclusive：是否是一个独占队列（一般等于true的话用于一个队列只能有一个消费者来消费的场景）
         *          有两个作用，一：当连接关闭时connection.close()该队列是否会自动删除；二：该队列是否是私有的private，
         *          如果不是排外的，可以使用两个消费者都访问同一个队列，没有任何问题，如果是排外的，会对当前队列加锁，其他通
         *          道channel是不能访问的，如果强制访问会报异常
         * autoDelete：是否自动删除，当最后一个消费者断开连接之后队列是否自动被删除
         * arguments：其他参数
         */
        channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null);
        String msg = null;
        long startms = System.currentTimeMillis();
        for( int i = 1; i<= 1000; i++) {
            // 初始化发送的消息
            msg = "单个发布确认信息：第" + i + "条";
            // 发送消息 这里要设置交换机名称，但是不设置路由
            // MessageProperties.PERSISTENT_TEXT_PLAIN：表示持久化
            channel.basicPublish("",Producer.QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes(StandardCharsets.UTF_8));
            boolean b = channel.waitForConfirms();
            if (!b) {
                System.out.println("第" + i + "条消息发送失败");
            }
        }
        long endms = System.currentTimeMillis();
        System.out.println("单个发布确认模式，发送1000条消息，消耗时间" + (endms - startms) + "ms");
        channel.close();

    }


    /**
     * 批量发布确认
     */
    public static void batchConfirm () throws Exception {

        // 创建信道
        Channel channel = RabbitFactory.createNewChannel();

        // 开启发布确认
        channel.confirmSelect();


        /*
         * 创建一个队列
         */
        channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null);
        String msg = null;
        long startms = System.currentTimeMillis();
        for( int i = 1; i<= 1000; i++) {
            // 初始化发送的消息
            msg = "批量发布确认信息：第" + i + "条";
            // 发送消息 这里要设置交换机名称，但是不设置路由
            // MessageProperties.PERSISTENT_TEXT_PLAIN：表示持久化
            channel.basicPublish("",Producer.QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes(StandardCharsets.UTF_8));
           // 满100条确认一次
            if (i % 100 == 0) {
                channel.waitForConfirms();
            }
        }
        channel.waitForConfirms();
        long endms = System.currentTimeMillis();
        System.out.println("批量发布确认模式，发送1000条消息，消耗时间" + (endms - startms) + "ms");
        channel.close();

    }

    /**
     * 异步发布确认
     */
    public static void asyncConfirm () throws Exception {

        // 创建信道
        Channel channel = RabbitFactory.createNewChannel();

        // 开启发布确认
        channel.confirmSelect();

        /*
         * 创建一个队列
         */
        channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null);

        ConfirmCallback ackCallback = (deliveryTag, multiple) ->{
//            System.out.println("消息发送失败：" +deliveryTag);
        };
        ConfirmCallback nackCallback = (deliveryTag, multiple) ->{
            System.out.println("消息发送失败：" +deliveryTag);

        };
        channel.addConfirmListener( ackCallback, nackCallback);

        String msg = null;
        long startms = System.currentTimeMillis();
        for( int i = 1; i<= 1000; i++) {
            // 初始化发送的消息
            msg = "异步发布确认信息：第" + i + "条";
            // 发送消息 这里要设置交换机名称，但是不设置路由
            // MessageProperties.PERSISTENT_TEXT_PLAIN：表示持久化
            channel.basicPublish("",Producer.QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes(StandardCharsets.UTF_8));
        }
        long endms = System.currentTimeMillis();
        System.out.println("异步发布确认模式，发送1000条消息，消耗时间" + (endms - startms) + "ms");
        channel.close();

    }
}
